package net.wlm.realtime.new_job;

import com.alibaba.fastjson.JSON;
import net.wlm.realtime.bean.VehicleData;
import net.wlm.realtime.utils.JdbcUtil;
import net.wlm.realtime.utils.KafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 行程识别与分析
 */

public class RealtimeVehicleTripRA {
    public static void main(String[] args) throws Exception {
        // 执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 数据源
        DataStream<String> vehicleData = KafkaUtil.consumerKafka(env, "vehicle-data");
        // 数据处理
        DataStream<VehicleData> vehicleStream = cleanData(vehicleData);
        // vehicleStream.print();
        // 行程分析处理   添加会话窗口计算


        // stringDataStream.print();
        // 数据输出

        // 执行
        env.execute();
    }


    private static DataStream<VehicleData> cleanData(DataStream<String> vehicleData) {
        SingleOutputStreamOperator<VehicleData> dataStream = vehicleData.filter(json -> {
                    try {
                        return json != null && JSON.isValid(json);
                    } catch (Exception e) {
                        return false;
                    }
                })
                // 将合法JSON字符串转换为VehicleData对象
                .map(json -> JSON.parseObject(json, VehicleData.class));
        return dataStream;
    }
}
